package kudu

import org.apache.kudu.client.KuduClient.KuduClientBuilder
import org.apache.kudu.client.{CreateTableOptions, KuduClient, SessionConfiguration}
import org.apache.kudu.{ColumnSchema, Schema, Type}
import org.junit.{Before, Test}

class KuduAPI {

  var kuduClient: KuduClient = null;

  // 初始化
  @Before
  def initKuduClient() = {
    // 1、创建Client
    val KUDU_MASTER = "s202:7051,s203:7051,s204:7051"
    kuduClient = new KuduClientBuilder(KUDU_MASTER).build()
  }

  @Test
  def createTable(): Unit = {

    // 2、创建schema
    val columns = List(
      new ColumnSchema.ColumnSchemaBuilder("id", Type.STRING).key(true).build(),
      new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).key(false).build()
    )
    // 隐式转换，
    import scala.collection.JavaConverters._
    val schema = new Schema(columns.asJava)

    // 3、创建分区
    val options = new CreateTableOptions()
      .setRangePartitionColumns(List("id").asJava)
      .setNumReplicas(1) // tablet是存储数据的位置，tablet可以有多个，有主从概念

    // 4、创建表
    kuduClient.createTable("kudutest", schema, options)
  }


  /**
   * 插入数据
   * 此处有个地方需要注意，如果在插入一条数相同id的数据，kudu API不会报错
   */
  @Test
  def insert(): Unit = {
    // 获取table, 表示对一个表进行操作
    val table = kuduClient.openTable("kudutest")

    // 获取一个操作， 组织对表数据的操作
    val insert = table.newInsert()
    val row = insert.getRow
    row.addString("id", "3")
    row.addString("name", "zs")

    // 开启会话，执行操作
    val session = kuduClient.newSession()
    session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC)
    session.apply(insert)
  }


  // 更新数据
  @Test
  def update(): Unit = {
    // 获取table, 表示对一个表进行操作
    val table = kuduClient.openTable("kudutest")

    // 获取一个操作， 组织对表数据的操作
    val update = table.newUpdate()
    val row = update.getRow
    row.addString("id", "3")
    row.addString("name", "zhangsan")

    // 开启会话，执行操作
    val session = kuduClient.newSession()
    session.apply(update)
  }

  // 删除数据
  @Test
  def delete(): Unit = {
    // 获取table, 表示对一个表进行操作
    val table = kuduClient.openTable("kudutest")

    // 获取一个操作， 组织对表数据的操作
    val delete = table.newDelete()
    val row = delete.getRow
    row.addString("id", "1")


    // 开启会话，执行操作
    val session = kuduClient.newSession()
    session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC)
    session.apply(delete)
  }

  @Test
  // 查询
  def query(): Unit = {
    // 1、构建一个扫描器
    var kuduScannerBuilder = kuduClient.newScannerBuilder(kuduClient.openTable("kudutest"))

    // 隐式转化，scala->java
    import scala.collection.JavaConverters._
    val columnNames = List("id", "name").asJava
    kuduScannerBuilder = kuduScannerBuilder.setProjectedColumnNames(columnNames)

    // 结果集
    val scanner = kuduScannerBuilder.build()
    while (scanner.hasMoreRows) {
      // 注意，这rows是获取一整个 tablet的数据
      val rows = scanner.nextRows()

      while (rows.hasNext) {
        val row = rows.next()
        println(row.getString(0), row.getString(1))
      }

    }
  }

  @Test
  def dropTable(): Unit = {
    val tableName = "kudutest2"
    if (kuduClient.tableExists(tableName)) kuduClient.deleteTable(tableName)
  }


}
